package com.bstek.bdf4.mq.publisher.service;

import com.bstek.bdf2.core.mq.publisher.BstekSendMessage;
import com.bstek.bdf2.core.mq.publisher.MessageInfo;
import com.bstek.bdf4.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.Resource;

/**
 * @Title: BstekSendMessageImpl
 * @Package: com.bstek.bdf4.mq.publisher.service
 * @Description:
 * @Author: zxq
 * @Date: 2024/2/5 - 14:39
 */

@Primary
@Service
@Slf4j
public class BstekSendMessageImpl implements BstekSendMessage {

    @Resource
    private RabbitTemplate rabbitTemplate;


    @Override
    public void send(MessageInfo messageInfo) {
        log.info("----------开始发送消息,MessageInfo:{}",messageInfo);
        rabbitTemplate.convertAndSend(messageInfo.getRoutingKey(),messageInfo.getMessage());
        rabbitTemplate.convertAndSend(messageInfo.getExchangeName(),messageInfo.getRoutingKey(),messageInfo.getMessage());

    }

    public void sendByCallBack(MessageInfo messageInfo) {
        log.info("----------开始发送消息,MessageInfo:{}",messageInfo);
        // 2.全局唯一的消息ID，需要封装到CorrelationData中
        CorrelationData correlationData = new CorrelationData(messageInfo.getMessageId());
        // 3.添加callback
        correlationData.getFuture().addCallback(
                result -> {
                    if(result.isAck()){
                        // 3.1.ack，消息成功
                        log.debug("消息发送成功, ID:{}", correlationData.getId());
                    }else{
                        // 3.2.nack，消息失败
                        log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                    }
                },
                ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
        );
         rabbitTemplate.convertAndSend(messageInfo.getExchangeName(),messageInfo.getRoutingKey(),messageInfo.getMessage(),correlationData);

    }

    public void sendTopicExchange(MessageInfo messageInfo){
        String exchangeName = messageInfo.getExchangeName();
        Assert.hasText(exchangeName,"交换机不能为空！");
        String topic = messageInfo.getTopic();
        Assert.hasText(topic,"topic不能为空！");
        Object message = messageInfo.getMessage();

        sendTopicExchange(exchangeName,topic,message);

    }

    /**
     * 根据交换机和topic( 通配符)发送消息
     * @param exchangeName 交换机
     * @param topic 通配符
     * @param message 消息
     */
     public void sendTopicExchange(String exchangeName,String topic,Object message){
         rabbitTemplate.convertAndSend(exchangeName,topic,message);
     }

}
